#include "stdafx.h"
#include "BlockingQueue.h"


BlockingQueue::BlockingQueue()
{
	this->capacity = 10;
	queue = new DataBuf[capacity];
	head = 0, tail = 0;
	pthread_mutex_init(&mutex, NULL);
	pthread_cond_init(&notFull, NULL);
	pthread_cond_init(&notEmpty, NULL);
}

BlockingQueue::BlockingQueue(int capacity)
{
	this->capacity = capacity;
	queue = new DataBuf[capacity];
	cout << "capacity " << sizeof(queue) << endl;
	head = 0, tail = 0;
	pthread_mutex_init(&mutex, NULL);
	pthread_cond_init(&notFull, NULL);
	pthread_cond_init(&notEmpty, NULL);

}


BlockingQueue::~BlockingQueue()
{
	this->capacity = 0;
	head = 0, tail = 0;
	delete queue;
	pthread_mutex_destroy(&mutex);
	pthread_cond_destroy(&notFull);
	pthread_cond_destroy(&notEmpty);
}

bool BlockingQueue::push(DataBuf item)
{
	pthread_mutex_lock(&mutex);
	while ((head + 1) % capacity == tail)//is full  
	{
		// push wait  
		pthread_cond_wait(&notFull, &mutex);
	}

	{
		queue[head] = item;
		head = (head + 1) % capacity;
		//wake up poll thread  
		pthread_cond_signal(&notEmpty);
		pthread_mutex_unlock(&mutex);

		return true;
	}
}


DataBuf BlockingQueue::poll()
{
	pthread_mutex_lock(&mutex);
	DataBuf ret;
	while (head == tail) // is empty  
	{
		//poll wait  
		pthread_cond_wait(&notEmpty, &mutex);
	}
	{
		ret = queue[tail];
		tail = (tail + 1) % capacity;
		//wake up push thread  
		pthread_cond_signal(&notFull);

		pthread_mutex_unlock(&mutex);
		return ret;
	}
}